package cluster

import (
	_ "sanguo/protocol/ss" //触发pb注册

	"github.com/golang/protobuf/proto"
	"github.com/sniperHW/kendynet"
	"github.com/sniperHW/kendynet/event"
	"github.com/sniperHW/kendynet/rpc"
	"github.com/sniperHW/kendynet/socket/stream_socket/tcp"
	//"sanguo/codec/ss"
	"fmt"
	"sanguo/codec/pb"
	"sanguo/common"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/sniperHW/kendynet/util"
)

//外部不能依赖PeerID的类型
type PeerID string

var selfService Service

var (
	queue   *event.EventQueue
	started int32
)

func (this PeerID) ToString() string {
	return string(this)
}

func MakePeerID(tt, ip string, port int32) PeerID {
	return MakeService(tt, ip, port).ToPeerID()
}

func makePeerID(str string) PeerID {
	return PeerID(str)
}

type Service struct {
	tt   string
	ip   string
	port int32
}

func (this Service) ToPeerID() PeerID {
	return PeerID(fmt.Sprintf("%s@%s:%d", this.tt, this.ip, this.port))
}

func MakeService(tt, ip string, port int32) Service {
	return Service{tt: tt, ip: ip, port: port}
}

func PeerID2Service(peerID PeerID) (Service, error) {

	s := Service{}

	p1 := strings.Split(peerID.ToString(), "@")
	if len(p1) != 2 {
		return s, fmt.Errorf("invaild peerID")
	}

	p2 := strings.Split(p1[1], ":")
	if len(p2) != 2 {
		return s, fmt.Errorf("invaild peerID")
	}

	s.tt = p1[0]
	s.ip = p2[0]

	port, err := strconv.Atoi(p2[1])
	if nil != err {
		return s, fmt.Errorf("invaild peerID")
	}
	s.port = int32(port)
	return s, nil
}

type rpcCall struct {
	arg       interface{}
	deadline  time.Time
	cb        rpc.RPCResponseHandler
}

func Brocast(tt string, msg proto.Message) {
	queue.PostNoWait(func() {
		defer mtx.Unlock()
		mtx.Lock()
		if ttmap, ok := ttEndPointMap[tt]; ok {
			for _, v := range ttmap.endPoints {
				PostMessage(v.toPeerID(), msg)
			}
		}
	})
}

/*
*  异步投递
 */
func PostMessage(peer PeerID, msg proto.Message) error {

	if started == 0 {
		return fmt.Errorf("cluster not started")
	}

	queue.PostNoWait(func() {
		endPoint := getEndPointByID(peer)
		if nil != endPoint {
			if nil != endPoint.conn {
				err := endPoint.conn.session.Send(msg)
				if nil != err {
					//记录日志
					Errorf("Send error:%s\n", err.Error())
				}
			} else {
				endPoint.pendingMsg = append(endPoint.pendingMsg, msg)
				//尝试与对端建立连接
				dial(endPoint)

			}
		} else {
			//记录日志
			Errorf("PostMessage %s not found", peer.ToString())
		}
	})

	return nil
}

var nextCheckHeartBeat int64

func tick() {
	now := time.Now().Unix()
	if now > nextCheckHeartBeat {
		nextCheckHeartBeat = now + 1
		for _, v := range idEndPointMap {
			if nil != v.conn && v.conn.nextHeartbeat != nil && now >= *v.conn.nextHeartbeat {
				*v.conn.nextHeartbeat = now + (common.HeartBeat_Timeout / 2)
				heartbeat := &Heartbeat{}
				heartbeat.Timestamp1 = proto.Int64(time.Now().UnixNano())
				v.conn.session.Send(heartbeat)
			}
		}
	}
	TickTimer()
}

/*
*  启动服务
 */
func Start(center_addr []string, def Service, bind_any ...bool) error {

	if !atomic.CompareAndSwapInt32(&started, 0, 1) {
		return fmt.Errorf("service already started")
	}

	selfService = def
	var ip string

	if len(bind_any) > 0 {
		ip = "0.0.0.0"
	} else {
		ip = def.ip
	}

	server, err := tcp.NewListener("tcp4", fmt.Sprintf("%s:%d", ip, def.port))
	if server != nil {

		connectCenter(center_addr, def)

		go func() {
			err := server.Start(func(session kendynet.StreamSession) {
				go func() {
					peerName, err := auth(session)
					if nil != err {
						session.Close(err.Error(), 0)
						return
					}

					service, err := PeerID2Service(PeerID(peerName))
					if nil != err {
						session.Close(err.Error(), 0)
						return
					}

					PostTask(func() {
						if !onEstablishServer(service, session) {
							session.Close("establish failed", 0)
							return
						}
					})

				}()
			})
			if nil != err {
				Errorf("server.Start() failed:%s\n", err.Error())
			}

		}()

		return nil
	} else {
		return err
	}
}

func GetEventQueue() *event.EventQueue {
	return queue
}

/*
*  将一个闭包投递到队列中执行，args为传递给闭包的参数
 */
func PostTask(function interface{}, args ...interface{}) {
	queue.PostNoWait(function, args...)
}

func init() {
	handlers = make(map[string]MsgHandler)
	queue = event.NewEventQueue()
	idEndPointMap = make(map[PeerID]*endPoint)
	ttEndPointMap = make(map[string]*typeEndPointMap)
	sessionPeerIDMap = make(map[kendynet.StreamSession]PeerID)
	center_handlers = make(map[string]MsgHandler)
	minheap = util.NewMinHeap(4096)
	nextCheckHeartBeat = time.Now().Unix()

	pb.Register("ss", &Heartbeat{}, 1)

	rpcServer, _ = rpc.NewRPCServer(&decoder{}, &encoder{})

	centerInit()

	rpc.InitClient(queue)

	go func() {
		for {
			queue.PostNoWait(tick)
			time.Sleep(time.Millisecond * 10)
		}
	}()

	go func() {
		queue.Run()
	}()

}
